package com.transaction.demo.message.service.impl;

import java.util.Date;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import lombok.extern.slf4j.Slf4j;

import com.transaction.demo.common.enums.PublicEnum;
import com.transaction.demo.common.util.JsonUtils;
import com.transaction.demo.common.util.ResultPage;
import com.transaction.demo.message.biz.TransactionMessageBiz;
import com.transaction.demo.message.config.MessageConfig;
import com.transaction.demo.message.enums.ExceptionEnum;
import com.transaction.demo.message.enums.MessageStatusEnum;
import com.transaction.demo.message.exception.MessageServiceException;
import com.transaction.demo.message.service.TransactionMessageService;
import com.transaction.demo.message.vo.req.TransactionMessageAddReqVo;
import com.transaction.demo.message.vo.req.TransactionMessageListReqVo;
import com.transaction.demo.message.vo.req.TransactionMessageUpdateReqVo;
import com.transaction.demo.message.vo.req.WaitOrSendMessageListReqVo;
import com.transaction.demo.message.vo.resp.TransactionMessageRespVo;

/**
 * 描述：
 * 作者：马庆智
 * 日期：2023/04/26 10:29:01
 */
@Slf4j
@Service
public class TransactionMessageServiceImpl implements TransactionMessageService {

	@Autowired
	private TransactionMessageBiz transactionMessageBiz;

	@Autowired
	private RocketMQTemplate rocketMQTemplate;

	@Autowired
	private MessageConfig config;

	/**
	 * 查询列表
	 *
	 * @param transactionMessage 请求参数
	 * @return TransactionMessageResp
	 * @author 马庆智
	 * @date: 2023/04/26 10:29:01
	 */
	@Override
	public ResultPage<TransactionMessageRespVo> queryTransactionMessageList(
			TransactionMessageListReqVo transactionMessage) {

		return transactionMessageBiz.queryTransactionMessageList(transactionMessage);
	}

	/**
	 * 预存储消息.
	 *
	 * @param message 消息
	 * @return id
	 * @author 马庆智
	 * @date: 2023/4/27 9:55
	 */
	@Override
	public int saveMessageWaitingConfirm(TransactionMessageAddReqVo message) {
		if (message == null) {
			throw new MessageServiceException(ExceptionEnum.MESSAGE_IS_NULL);
		}

		if (StringUtils.isEmpty(message.getConsumerQueue())) {
			throw new MessageServiceException(ExceptionEnum.MESSAGE_CONSUMER_QUEUE_IS_NULL);
		}

		message.setCreateTime(new Date());
		message.setStatus(MessageStatusEnum.WAITING_CONFIRM.name());
		message.setAreadlyDead(PublicEnum.NO.name());
		message.setMessageSendTimes((short) 0);

		return transactionMessageBiz.saveTransactionMessage(message);
	}

	/**
	 * 确认并发送消息.
	 *
	 * @param messageId 消息id
	 * @author 马庆智
	 * @date: 2023/4/27 10:01
	 */
	@Override
	public int confirmAndSendMessage(String messageId) {
		TransactionMessageRespVo message = transactionMessageBiz.queryTransactionMessage(messageId);
		if (message == null) {
			throw new MessageServiceException(ExceptionEnum.MESSAGE_IS_NULL);
		}

		message.setStatus(MessageStatusEnum.SENDING.name());
		message.setUpdateTime(new Date());

		TransactionMessageUpdateReqVo updateReqVo = new TransactionMessageUpdateReqVo();
		BeanUtils.copyProperties(message, updateReqVo);
		transactionMessageBiz.updateTransactionMessage(updateReqVo);

		final SendResult result = rocketMQTemplate.syncSend(message.getConsumerQueue(), message.getMessageBody());
		log.info("确认并发送消息响应:{}", JsonUtils.toJson(result));
		return 1;
	}

	/**
	 * 存储并发送消息.
	 *
	 * @param message 消息
	 * @return id
	 * @author 马庆智
	 * @date: 2023/4/27 9:55
	 */
	@Override
	public int saveAndSendMessage(TransactionMessageAddReqVo message) {
		log.info("存储并发送消息参数:{}", JsonUtils.toJson(message));
		if (message == null) {
			throw new MessageServiceException(ExceptionEnum.MESSAGE_IS_NULL);
		}

		if (StringUtils.isEmpty(message.getConsumerQueue())) {
			throw new MessageServiceException(ExceptionEnum.MESSAGE_CONSUMER_QUEUE_IS_NULL);
		}

		message.setStatus(MessageStatusEnum.SENDING.name());
		message.setAreadlyDead(PublicEnum.NO.name());
		message.setMessageSendTimes((short) 0);
		message.setUpdateTime(new Date());

		int id = transactionMessageBiz.saveTransactionMessage(message);

		SendResult result = rocketMQTemplate.syncSend(message.getConsumerQueue(), message.getMessageBody());
		log.info("存储并发送消息响应:{}", JsonUtils.toJson(result));
		return id;
	}

	/**
	 * 直接发送消息.
	 *
	 * @param message 消息
	 * @author 马庆智
	 * @date: 2023/4/27 9:55
	 */
	@Override
	public int directSendMessage(TransactionMessageAddReqVo message) {
		if (message == null) {
			throw new MessageServiceException(ExceptionEnum.MESSAGE_IS_NULL);
		}

		if (StringUtils.isEmpty(message.getConsumerQueue())) {
			throw new MessageServiceException(ExceptionEnum.MESSAGE_CONSUMER_QUEUE_IS_NULL);
		}

		short times = message.getMessageSendTimes();

		log.info("直接发送消息次数:{}", times);
		SendResult result = rocketMQTemplate.syncSend(message.getConsumerQueue(),
				MessageBuilder.withPayload(message.getMessageBody()).build(), config.timeout, times);

		log.info("直接发送消息参数：{},响应:{}", JsonUtils.toJson(message), JsonUtils.toJson(result));
		return 1;
	}

	/**
	 * 重发消息.
	 *
	 * @param message 消息
	 * @author 马庆智
	 * @date: 2023/4/27 9:55
	 */
	@Override
	public int reSendMessage(TransactionMessageAddReqVo message) {
		log.info("重发消息: {}", JsonUtils.toJson(message));
		if (message == null) {
			throw new MessageServiceException(ExceptionEnum.MESSAGE_IS_NULL);
		}

		if (StringUtils.isEmpty(message.getConsumerQueue())) {
			throw new MessageServiceException(ExceptionEnum.MESSAGE_CONSUMER_QUEUE_IS_NULL);
		}

		if (message.getMessageSendTimes() >= config.maxTimes) {
			message.setAreadlyDead(PublicEnum.YES.name());
		}

		message.setMessageSendTimes((short) (message.getMessageSendTimes() + (short) 1));
		message.setUpdateTime(new Date());

		TransactionMessageUpdateReqVo updateReqVo = new TransactionMessageUpdateReqVo();
		BeanUtils.copyProperties(message, updateReqVo);
		log.info("更新消息发送的次数和时间: {}", JsonUtils.toJson(updateReqVo));
		transactionMessageBiz.updateTransactionMessage(updateReqVo);

		log.info("向消息队列：{},投递消息: {}", message.getConsumerQueue(), message.getMessageBody());
		final SendResult result = rocketMQTemplate.syncSend(message.getConsumerQueue(), message.getMessageBody());
		log.info("重发消息响应:{}", JsonUtils.toJson(result));
		return 0;
	}

	/**
	 * 根据messageId重发某条消息.
	 *
	 * @param messageId 消息id
	 * @author 马庆智
	 * @date: 2023/4/27 9:55
	 */
	@Override
	public int reSendMessageByMessageId(String messageId) {
		TransactionMessageRespVo message = transactionMessageBiz.queryTransactionMessage(messageId);

		TransactionMessageAddReqVo addReqVo = new TransactionMessageAddReqVo();
		BeanUtils.copyProperties(message, addReqVo);

		return this.reSendMessage(addReqVo);
	}

	/**
	 * 将消息标记为死亡消息.
	 *
	 * @param messageId 消息id
	 * @author 马庆智
	 * @date: 2023/4/27 9:55
	 */
	@Override
	public int setMessageToAreadlyDead(String messageId) {
		TransactionMessageRespVo message = transactionMessageBiz.queryTransactionMessage(messageId);
		if (message == null) {
			throw new MessageServiceException(ExceptionEnum.MESSAGE_IS_NULL);
		}

		message.setAreadlyDead(PublicEnum.YES.name());
		message.setUpdateTime(new Date());

		TransactionMessageUpdateReqVo updateReqVo = new TransactionMessageUpdateReqVo();
		BeanUtils.copyProperties(message, updateReqVo);

		return transactionMessageBiz.updateTransactionMessage(updateReqVo);
	}

	/**
	 * 根据消息ID获取消息
	 *
	 * @param messageId 消息id
	 * @return TransactionMessageAddReqVo
	 * @author 马庆智
	 * @date: 2023/4/27 9:55
	 */
	@Override
	public TransactionMessageRespVo getMessageByMessageId(String messageId) {
		return transactionMessageBiz.queryTransactionMessage(messageId);
	}

	/**
	 * 根据消息ID删除消息
	 *
	 * @param messageId 消息id
	 * @author 马庆智
	 * @date: 2023/4/27 9:55
	 */
	@Override
	public int deleteMessageByMessageId(String messageId) {
		return transactionMessageBiz.deleteTransactionMessage(messageId);
	}

	/**
	 * 重发某个消息队列中的全部已死亡的消息.
	 *
	 * @param queueName 队列名称
	 * @param batchSize 发送的数量
	 * @author 马庆智
	 * @date: 2023/4/27 9:55
	 */
	@Override
	public long reSendAllDeadMessageByQueueName(String queueName, int batchSize) {
		int current = 1;
		TransactionMessageListReqVo msg = new TransactionMessageListReqVo();
		msg.setConsumerQueue(queueName);
		msg.setAreadlyDead(PublicEnum.YES.name());
		msg.setCurrent(current);
		msg.setPageSize(batchSize);

		ResultPage<TransactionMessageRespVo> messageList = transactionMessageBiz.queryTransactionMessageList(msg);

		long total = messageList.getTotal();
		//总页数
		long l = total / batchSize + 1;

		log.info("开始重发消息队列{}中的全部已死亡的消息,总数：{}", queueName, total);

		while (current <= l) {
			for (TransactionMessageRespVo message : messageList.getList()) {
				TransactionMessageAddReqVo addReqVo = new TransactionMessageAddReqVo();
				BeanUtils.copyProperties(message, addReqVo);
				this.reSendMessage(addReqVo);
			}
			current++;
		}

		log.info("结束重发消息队列{}中的全部已死亡的消息,总数：{}", queueName, total);
		return total;
	}

	/**
	 * 查询待确认或者发送中的的消息
	 *
	 * @param record 请求参数
	 * @return TransactionMessageResp
	 * @author 马庆智
	 * @date: 2023/04/26 10:29:01
	 */
	@Override
	public ResultPage<TransactionMessageRespVo> queryWaitOrSendMessageList(WaitOrSendMessageListReqVo record) {
		return transactionMessageBiz.queryWaitOrSendMessageList(record);
	}
}